package org.research;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.Topic;
import javax.naming.InitialContext;
import javax.naming.NamingException;

import org.apache.log4j.Logger;

public class Consumer {

    private static final Logger log = Logger.getLogger(Consumer.class);

    private ConnectionFactory connectionFactory;
    private Connection connection;
    private Session session;
    private Listener listener;
    private Topic topic;
    private MessageConsumer consumer;

    public Consumer(Listener listener) {
        this.listener = listener;
        try {
            InitialContext jndiContext = new InitialContext();
            connectionFactory = (ConnectionFactory) jndiContext.lookup("ConnectionFactory");
            log.debug(connectionFactory);
            connection = connectionFactory.createConnection();
            log.debug("JMS connection opened " + connection);
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            log.debug("JMS session created " + session);
            topic = (Topic) jndiContext.lookup("MyTopic");
            log.debug("Looked up topic: " + topic);
            consumer = session.createConsumer(topic);
            consumer.setMessageListener(new MessageListener() {
                public void onMessage(Message message) {
                    try {
                        Consumer.this.listener.execute(message);
                    }
                    catch (JMSException e) {
                        log.error("Unable to execute listener", e);
                    }
                }
            });
        }
        catch (NamingException e) {
            throw new IllegalStateException("Unable to resolve jndi name", e);
        }
        catch (JMSException e) {
            throw new IllegalStateException("Unable to create connection", e);
        }
    }

    public static void start(Listener listener) {
        Consumer consumer = new Consumer(listener);
        consumer.startReceiveMessagesAsynchronously();
    }

    private void startReceiveMessagesAsynchronously() {
        try {
            connection.start();
            log.debug("Start receiving messages");
        }
        catch (JMSException e) {
            throw new IllegalStateException("Unable to start connection", e);
        }
    }

    public static interface Listener {
        void execute(Message message) throws JMSException;
    }
}
